package com.shujia.wyh.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/*
    包装过滤器
        SkipFilter  跳过符合查询条件的数据
        WhileMatchFilter 直到符合查询条件的数据就停止
 */
public class HbaseFilterDemo3 {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        try {
            //获取与Hbase的连接对象
            //1、获取Hadoop的相关环境
            Configuration conf = new Configuration();

            //2、设置以下zookeeper的集群信息
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //3、获取habse的连接对象
            conn = ConnectionFactory.createConnection(conf);

            //4、获取HMaster的对象
            admin = conn.getAdmin();

            System.out.println("获取连接成功！！" + conn);
            System.out.println("HMaster的对象获取成功！！" + admin);
            System.out.println("------------------------------------------------------------------");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * ValueFilter如果不等值查询,结果只会没有那一列,但是同属于一行的其他列数据会被查出来
     * SkipFilter再包装一下的话,同属于一行的所有列都不会被查询
     */
    @Test
    public void skipFilterFun() {
        try {
            try {
                //获取表的实例
                //先将表封装成TableName的对象
                TableName students2 = TableName.valueOf("students2");

                //获取表的实例对象
                Table table = conn.getTable(students2);

                Scan scan = new Scan();

//                SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
//                        Bytes.toBytes("shujia"),
//                        Bytes.toBytes("性别"),
//                        CompareOperator.EQUAL,
//                        new SubstringComparator("男"));

                Filter filter1 = new ValueFilter(CompareOperator.NOT_EQUAL,new BinaryComparator(Bytes.toBytes("羿旭炎")));

                //创建包装过滤器
                SkipFilter skipFilter = new SkipFilter(filter1);

                scan.setFilter(skipFilter);

                //查询数据
                ResultScanner resultScanner = table.getScanner(scan);

                printResultScanner(resultScanner);


            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *  WhileMatchFilter 当遇到某一行的时候停止
     */
    @Test
    public void whileMatchFilterFun(){
        try {
            try {
                //获取表的实例
                //先将表封装成TableName的对象
                TableName students2 = TableName.valueOf("students2");

                //获取表的实例对象
                Table table = conn.getTable(students2);

                Scan scan = new Scan();


                RowFilter rowFilter = new RowFilter(CompareOperator.NOT_EQUAL, new BinaryPrefixComparator(Bytes.toBytes("1500100014")));

                WhileMatchFilter whileMatchFilter = new WhileMatchFilter(rowFilter);

                scan.setFilter(whileMatchFilter);

                //查询数据
                ResultScanner resultScanner = table.getScanner(scan);

                printResultScanner(resultScanner);


            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过运用4种比较器过滤出姓于，年纪小于等于23岁，性别为女，且是理科的学生。
     *
     * singleColumnValueFilter
     *
     *  BinaryPrefixComparator姓于
     *  BinaryComparator   小于等于23岁
     *  SubstringComparator  性别为女
     *  RegexStringComparator 理科的学生
     */
    @Test
    public void zuHeFilter(){
        try {
            try {
                //获取表的实例
                //先将表封装成TableName的对象
                TableName students2 = TableName.valueOf("students2");

                //获取表的实例对象
                Table table = conn.getTable(students2);

                Scan scan = new Scan();

                SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                        Bytes.toBytes("info"),
                        Bytes.toBytes("姓名"),
                        CompareOperator.EQUAL,
                        new BinaryPrefixComparator(Bytes.toBytes("于"))
                );

                SingleColumnValueFilter filter2 = new SingleColumnValueFilter(
                        Bytes.toBytes("info"),
                        Bytes.toBytes("年龄"),
                        CompareOperator.LESS_OR_EQUAL,
                        new BinaryComparator(Bytes.toBytes("23"))
                );

                SingleColumnValueFilter filter3 = new SingleColumnValueFilter(
                        Bytes.toBytes("shujia"),
                        Bytes.toBytes("性别"),
                        CompareOperator.EQUAL,
                        new SubstringComparator("女")
                );

                SingleColumnValueFilter filter4 = new SingleColumnValueFilter(
                        Bytes.toBytes("shujia"),
                        Bytes.toBytes("班级"),
                        CompareOperator.EQUAL,
                        new RegexStringComparator("^理科")
                );

                /*scan.setFilter(filter1);
                scan.setFilter(filter2);
                scan.setFilter(filter3);
                scan.setFilter(filter4);*/

                //创建一个过滤器列表
                FilterList filterList = new FilterList();
                //将每一个过滤器添加到列表中
                filterList.addFilter(filter1);
                filterList.addFilter(filter2);
                filterList.addFilter(filter3);
                filterList.addFilter(filter4);


                scan.setFilter(filterList);


                //查询数据
                ResultScanner resultScanner = table.getScanner(scan);

                printResultScanner(resultScanner);


            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *  过滤出学号是以15001001开头的文科学生
     *
     *  1 行键过滤器
     *  2 单列值过滤器
     */
    @Test
    public void zuHeFilter2Fun(){
        try {
            try {
                //获取表的实例
                //先将表封装成TableName的对象
                TableName students2 = TableName.valueOf("students2");

                //获取表的实例对象
                Table table = conn.getTable(students2);

                Scan scan = new Scan();

                //创建一个行键过滤器过滤行键值以15001001开头
                RowFilter filter2 = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("15001001")));

                SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                        Bytes.toBytes("shujia"),
                        Bytes.toBytes("班级"),
                        CompareOperator.EQUAL,
                        new BinaryPrefixComparator(Bytes.toBytes("文科"))
                );


                //创建一个过滤器列表
                FilterList filterList = new FilterList();
                //将每一个过滤器添加到列表中
                filterList.addFilter(filter1);
                filterList.addFilter(filter2);
                scan.setFilter(filterList);


                //查询数据
                ResultScanner resultScanner = table.getScanner(scan);

                printResultScanner(resultScanner);


            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }




    @After
    public void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 自定义一个方法输出getScanner的结果值
     */
    public void printResultScanner(ResultScanner resultScanner) {
        try {
            //定义一个变量是Result类型的
            Result rs = null;
            //将每一次获取的值赋值给rs,如果rs不为null,说明获取到值
            //rs获取的是一行的值，而一行可以有若干个列
            while ((rs = resultScanner.next()) != null) {
                //获取到行键值
                String id = Bytes.toString(rs.getRow());
                //上面的获取，是在我们已经知道了列名的前提下获取的，如果我们不知道列名，又怎么获取呢？
                //获取一行中所有的列，每一列组成了Cell对象，返回一个集合
                List<Cell> cells = rs.listCells();

                StringBuilder sb = new StringBuilder();
                sb.append("学号：").append(id);

                for (Cell cell : cells) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    sb.append(", ").append(qualifier).append("：").append(value);
                }

                //得到一行的拼接结果
                System.out.println(sb);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
